package server_timu;

import com.google.gson.Gson;
import com.google.gson.JsonElement;
import com.google.gson.JsonObject;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.connector.kafka.sink.KafkaRecordSerializationSchema;
import org.apache.flink.connector.kafka.sink.KafkaSink;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SideOutputDataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.RichAllWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;
import org.apache.kafka.clients.consumer.OffsetResetStrategy;

import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.Comparator;
import java.util.List;
import java.util.stream.Collectors;
import java.util.stream.Stream;
import java.util.stream.StreamSupport;

import static util.method_util.format_time;

/*
      1、使用Flink消费Kafka中topic为ods_mall_data的数据，根据数据中不同的表将数据分别分发至kafka的
      DWD层的fact_order_master、fact_order_detail的Topic中（只获取data的内容，具体的内容格式考生
      请自查），其他的表则无需处理，其中请对进入到fact_order_master的内容进行排序，同时迟到容忍度设置
      为2分钟；

      todo 不知道为啥，用gson的话会包一个序列化的问题,虽然报错是说SimpleDateFormat的错，但是我换成flink自带的
      todo 解析json数据的包就不会报这个错
 */
public class T1_gson {
    public static void main(String[] args) throws Exception {

        //  todo 创建流式环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        //  todo 创建Gson对象，实例化   处理json数据
        Gson gson = new Gson();

        //  todo 创建测输出流存储order_detail的数据  这里的写法避免了泛型擦除
        OutputTag<String> detail_out = new OutputTag<String>("detail_out"){};



        //  todo 拿到topic_db主题的数据

        //  创建kafkasource
        //  todo setStartingOffsets:用于指定 KafkaSource 启动时从哪个位置开始读取数据。
        //  todo OffsetsInitializer.committedOffsets(OffsetResetStrategy.LATEST):
        //  todo 表示从 Kafka 中该消费者组（groupId）已提交的偏移量（committed offsets）开始消费。
        KafkaSource<String> kafka_source = KafkaSource.<String>builder()
                .setBootstrapServers("192.168.40.110:9092")
                .setGroupId("ods_mall_data_group")
                .setValueOnlyDeserializer(new SimpleStringSchema())                 //  只对值进行反序列化
                .setTopics("topic_db")
                .setStartingOffsets(OffsetsInitializer.committedOffsets(OffsetResetStrategy.LATEST))
                .build();

        //  得到数据
        DataStreamSource<String> data = env.fromSource(
                kafka_source,
                WatermarkStrategy.noWatermarks(),
                "kafka_source"
        );


        data.print("查看源数据格式");

        //  todo 这里使用gson的话返回的数据是JsonElement类型的数据
        SingleOutputStreamOperator<JsonElement> info_data = data.process(new ProcessFunction<String, JsonElement>() {
            @Override
            public void processElement(String value, ProcessFunction<String, JsonElement>.Context context, Collector<JsonElement> collector) throws Exception {
                //  todo 将得到的数据解析成gson的JsonElement对象
                JsonElement jsonElement = gson.fromJson(value, JsonElement.class);
                //  todo 将JsonElement对象转换成JsonObject对象
                JsonObject json_object = jsonElement.getAsJsonObject();

                //  todo 拿到数据的表名
                String table = json_object.get("table").getAsString();
                //  todo 拿到data的内容
                JsonElement data = json_object.get("data");

                if (table.contains("order_master")) {
                    //  todo 向下游推送order_master的数据
                    collector.collect(data);
                } else if (table.contains("order_detail")) {
                    //  todo order_detail数据放到测输出流
                    context.output(detail_out, data.toString());
                }
            }
        });


        //  todo 对主流数据进行排序(上面已经将数据流分开)
        //  todo WindowAll:全局窗口，将所有数据放到一个窗口，即不分区，分组，并行度为1
        SingleOutputStreamOperator<String> master_data = info_data.windowAll(TumblingProcessingTimeWindows.of(Time.seconds(20)))
                .allowedLateness(Time.minutes(2))
                //  todo 第一个参数是输入的数据类型，中间的是返回的数据类型，第三个是窗口的类型
                //  应用一个全窗口函数
                .apply(new RichAllWindowFunction<JsonElement, String, TimeWindow>() {

                    // 静态内部类中持有ThreadLocal，确保其不依赖外部实例
                    private transient SimpleDateFormat format;

                    @Override
                    public void open(Configuration parameters) {
                        format = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss"); // 在 open 中初始化

                    }

                    @Override
                    public void apply(TimeWindow window, Iterable<JsonElement> values, Collector<String> collect) throws Exception {

                        //  将Iterable的数据转化为java的Stream类型的数据，方便使用流式操作
                        //  false：表示不并行处理（单线程顺序执行）。
                        //  values.spliterator()：将 Iterable 转换为可分割的迭代器（Spliterator）。
                        Stream<JsonElement> stream = StreamSupport.stream(values.spliterator(), false);
                        //  对流式数据进行排序,按照modified_time字段，并且格式化后的数据
                        //  todo Comparator.comparing方法：创建一个比较器，用于比较两个对象。
                        //  todo collect(Collectors.toList()):将排序后的 Stream<JsonElement> 收集为 List<JsonElement>。
                        //  Collectors:收集器
                        List<JsonElement> sort_list = stream.sorted(Comparator.comparing(strings -> {
                            //  todo 由于format_date.parse()需要抛出一个parse的异常
                            try {
                                //   todo 时间格式化
//                                SimpleDateFormat format = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
                                //  todo format.parse(字符串).getTime得到格式化的时间
                                //  todo 只有JsonObject才有get方法，jsonElement是没有的
                                return format_time(strings.getAsJsonObject().get("modified_time").getAsString());
                            } catch (ParseException e) {
                                throw new RuntimeException(e);
                            }
                        })).collect(Collectors.toList());
                        //  todo 循环将每一条数据向下游发送
                        for (JsonElement json : sort_list) {
                            collect.collect(json.toString());
                        }
                    }
                });

        //  todo 拿到测输出流的数据
        SideOutputDataStream<String> detail_data = info_data.getSideOutput(detail_out);

        //  todo 根据题目创建不同的kafka_sink
        KafkaSink<String> kafkaSink_master = KafkaSink.<String>builder()
                .setBootstrapServers("192.168.40.110:9092")
                .setRecordSerializer(
                        KafkaRecordSerializationSchema.<String>builder()
                                .setTopic("fact_order_master")
                                .setValueSerializationSchema(new SimpleStringSchema())
                                .build()
                )
                .build();

        KafkaSink<String> kafkaSink_detail = KafkaSink.<String>builder()
                .setBootstrapServers("192.168.40.110:9092")
                .setRecordSerializer(
                        KafkaRecordSerializationSchema.<String>builder()
                                .setTopic("fact_order_detail")
                                .setValueSerializationSchema(new SimpleStringSchema())
                                .build()

                )
                .build();


        //  todo 数据写入对应kafka主题
        master_data.sinkTo(kafkaSink_master);
        detail_data.sinkTo(kafkaSink_detail);

        //  todo 将主流数据输出到error错误流这样是为了区分两条数据流
        master_data.printToErr();
        detail_data.print();


        env.execute();




    }
}
